package com.zhengb.flink.study.datastream.transformation;

import com.alibaba.fastjson.JSONObject;
import com.zhengb.flink.study.datastream.datasource.DriverLocationSource;
import com.zhengb.flink.study.datastream.entity.LocationEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * 自定义非并行 source function
 * <p>
 * DataStream -> KeyedStream
 *
 * @author zhengbo
 * @date 2019/12/19
 */
@Slf4j
public class KeyByTransformation {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.addSource(new DriverLocationSource(), "driverSource").setParallelism(10)
                .keyBy(new KeySelector<LocationEntity, String>() {
                    //相同司机id的数据会被分配到同一个sink线程中
                    @Override
                    public String getKey(LocationEntity value) throws Exception {
                        return value.getDriverNo();
                    }
                }).
                addSink(new RichSinkFunction<LocationEntity>() {
                    @Override
                    public void invoke(LocationEntity value, Context context) throws Exception {
                        log.info("receive driver data:{},{}", value.getDriverNo(), JSONObject.toJSONString(value));
                    }
                }).setParallelism(6);


        env.execute("keyByTransformation");
    }


}
